//! S3对象存储数据源插件
//!
//! 提供对Amazon S3和兼容S3的对象存储的访问支持

use async_trait::async_trait;
use std::collections::HashMap;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use crate::{DataSource, DataSourceQuery, DataSourceResult, DataSourceStats};

/// S3配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct S3Config {
    /// 端点URL
    pub endpoint: String,
    /// 访问密钥ID
    pub access_key_id: String,
    /// 秘密访问密钥
    pub secret_access_key: String,
    /// 区域
    pub region: String,
    /// 存储桶名称
    pub bucket: String,
    /// 路径前缀
    pub prefix: Option<String>,
}

impl Default for S3Config {
    fn default() -> Self {
        Self {
            endpoint: "https://s3.amazonaws.com".to_string(),
            access_key_id: "".to_string(),
            secret_access_key: "".to_string(),
            region: "us-east-1".to_string(),
            bucket: "".to_string(),
            prefix: None,
        }
    }
}

/// S3对象信息
#[derive(Debug, Clone)]
pub struct S3Object {
    /// 对象键
    pub key: String,
    /// 大小（字节）
    pub size: i64,
    /// 最后修改时间
    pub last_modified: String,
    /// ETag
    pub etag: String,
    /// 存储类别
    pub storage_class: String,
}

/// S3数据源
pub struct S3DataSource {
    /// 数据源名称
    name: String,
    /// 数据源描述
    description: String,
    /// S3配置
    config: S3Config,
    /// 连接状态
    connected: bool,
    /// 缓存的对象列表
    objects_cache: Vec<S3Object>,
}

impl S3DataSource {
    /// 创建新的S3数据源
    pub fn new(config: S3Config) -> Self {
        Self {
            name: "S3".to_string(),
            description: "Amazon S3对象存储数据源".to_string(),
            config,
            connected: false,
            objects_cache: Vec::new(),
        }
    }

    /// 列出S3对象（模拟实现）
    fn list_objects(&mut self) -> anyhow::Result<Vec<S3Object>> {
        // 简化实现：返回模拟的对象列表
        let objects = vec![
            S3Object {
                key: "data/users.csv".to_string(),
                size: 1024,
                last_modified: "2024-01-01T00:00:00Z".to_string(),
                etag: "\"abc123\"".to_string(),
                storage_class: "STANDARD".to_string(),
            },
            S3Object {
                key: "data/orders.json".to_string(),
                size: 2048,
                last_modified: "2024-01-02T00:00:00Z".to_string(),
                etag: "\"def456\"".to_string(),
                storage_class: "STANDARD".to_string(),
            },
            S3Object {
                key: "logs/app.log".to_string(),
                size: 4096,
                last_modified: "2024-01-03T00:00:00Z".to_string(),
                etag: "\"ghi789\"".to_string(),
                storage_class: "STANDARD".to_string(),
            },
        ];

        // 更新缓存
        self.objects_cache = objects.clone();

        Ok(objects)
    }

    /// 获取对象内容（模拟实现）
    fn get_object(&self, key: &str) -> anyhow::Result<String> {
        // 简化实现：返回模拟的对象内容
        match key {
            "data/users.csv" => Ok("id,name,email\n1,Alice,alice@example.com\n2,Bob,bob@example.com".to_string()),
            "data/orders.json" => Ok(r#"[{"id":1,"user_id":1,"amount":100.0},{"id":2,"user_id":2,"amount":200.0}]"#.to_string()),
            "logs/app.log" => Ok("2024-01-01 INFO: Application started\n2024-01-01 INFO: Processing request".to_string()),
            _ => Err(anyhow::anyhow!("对象未找到: {}", key)),
        }
    }

    /// 将S3对象列表转换为JSON
    fn objects_to_json(&self, objects: Vec<S3Object>) -> Value {
        let objects_json: Vec<Value> = objects.iter().map(|obj| {
            json!({
                "key": obj.key,
                "size": obj.size,
                "last_modified": obj.last_modified,
                "etag": obj.etag,
                "storage_class": obj.storage_class
            })
        }).collect();

        json!(objects_json)
    }

    /// 解析简单的SQL查询（仅支持基本的SELECT）
    fn parse_sql(&self, sql: &str) -> anyhow::Result<String> {
        let sql_lower = sql.to_lowercase();

        if sql_lower.contains("select") && sql_lower.contains("from") {
            if sql_lower.contains("objects") || sql_lower.contains("s3") {
                Ok("list_objects".to_string())
            } else {
                Err(anyhow::anyhow!("不支持的表名"))
            }
        } else {
            Err(anyhow::anyhow!("仅支持SELECT查询"))
        }
    }
}

#[async_trait]
impl DataSource for S3DataSource {
    fn name(&self) -> &str {
        &self.name
    }

    fn description(&self) -> &str {
        &self.description
    }

    async fn connect(&mut self) -> anyhow::Result<()> {
        // 简化实现：模拟连接过程
        println!("正在连接到S3: endpoint={}, bucket={}", self.config.endpoint, self.config.bucket);

        // 模拟连接延迟
        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;

        // 标记为已连接
        self.connected = true;

        println!("S3连接成功");
        Ok(())
    }

    async fn disconnect(&mut self) -> anyhow::Result<()> {
        self.connected = false;
        println!("S3连接已断开");
        Ok(())
    }

    async fn is_connected(&self) -> bool {
        self.connected
    }

    async fn execute_query(&mut self, query: DataSourceQuery) -> anyhow::Result<DataSourceResult> {
        let start_time = std::time::Instant::now();

        if !self.is_connected().await {
            return Ok(DataSourceResult {
                success: false,
                data: json!(null),
                error: Some("S3未连接".to_string()),
                stats: DataSourceStats {
                    execution_time_ms: start_time.elapsed().as_millis() as f64,
                    rows_processed: 0,
                    bytes_processed: 0,
                },
            });
        }

        // 解析SQL查询
        let operation = match self.parse_sql(&query.content) {
            Ok(op) => op,
            Err(e) => {
                return Ok(DataSourceResult {
                    success: false,
                    data: json!(null),
                    error: Some(e.to_string()),
                    stats: DataSourceStats {
                        execution_time_ms: start_time.elapsed().as_millis() as f64,
                        rows_processed: 0,
                        bytes_processed: 0,
                    },
                });
            }
        };

        match operation.as_str() {
            "list_objects" => {
                match self.list_objects() {
                    Ok(objects) => {
                        let data = self.objects_to_json(objects.clone());
                        Ok(DataSourceResult {
                            success: true,
                            data,
                            error: None,
                            stats: DataSourceStats {
                                execution_time_ms: start_time.elapsed().as_millis() as f64,
                                rows_processed: objects.len() as u64,
                                bytes_processed: 0, // 简化实现
                            },
                        })
                    }
                    Err(e) => {
                        Ok(DataSourceResult {
                            success: false,
                            data: json!(null),
                            error: Some(e.to_string()),
                            stats: DataSourceStats {
                                execution_time_ms: start_time.elapsed().as_millis() as f64,
                                rows_processed: 0,
                                bytes_processed: 0,
                            },
                        })
                    }
                }
            }
            _ => {
                Ok(DataSourceResult {
                    success: false,
                    data: json!(null),
                    error: Some(format!("不支持的操作: {}", operation)),
                    stats: DataSourceStats {
                        execution_time_ms: start_time.elapsed().as_millis() as f64,
                        rows_processed: 0,
                        bytes_processed: 0,
                    },
                })
            }
        }
    }

    async fn health_check(&self) -> anyhow::Result<Value> {
        Ok(json!({
            "status": if self.is_connected().await { "connected" } else { "disconnected" },
            "endpoint": self.config.endpoint,
            "bucket": self.config.bucket,
            "objects_cached": self.objects_cache.len()
        }))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn create_test_s3_config() -> S3Config {
        S3Config {
            endpoint: "https://s3.amazonaws.com".to_string(),
            access_key_id: "test_access_key".to_string(),
            secret_access_key: "test_secret_key".to_string(),
            region: "us-east-1".to_string(),
            bucket: "test-bucket".to_string(),
            prefix: Some("data/".to_string()),
        }
    }

    #[tokio::test]
    async fn test_s3_data_source_creation() {
        let config = create_test_s3_config();
        let s3_source = S3DataSource::new(config);

        assert_eq!(s3_source.name(), "S3");
        assert_eq!(s3_source.description(), "Amazon S3对象存储数据源");
        assert!(!s3_source.is_connected().await);
    }

    #[tokio::test]
    async fn test_s3_connection_lifecycle() {
        let config = create_test_s3_config();
        let mut s3_source = S3DataSource::new(config);

        // 初始状态应该是未连接
        assert!(!s3_source.is_connected().await);

        // 连接
        s3_source.connect().await.unwrap();
        assert!(s3_source.is_connected().await);

        // 断开连接
        s3_source.disconnect().await.unwrap();
        assert!(!s3_source.is_connected().await);
    }

    #[tokio::test]
    async fn test_s3_list_objects() {
        let config = create_test_s3_config();
        let mut s3_source = S3DataSource::new(config);

        s3_source.connect().await.unwrap();

        let objects = s3_source.list_objects().unwrap();
        assert_eq!(objects.len(), 3);
        assert_eq!(objects[0].key, "data/users.csv");
        assert_eq!(objects[1].key, "data/orders.json");
        assert_eq!(objects[2].key, "logs/app.log");
    }

    #[tokio::test]
    async fn test_s3_get_object() {
        let config = create_test_s3_config();
        let mut s3_source = S3DataSource::new(config);

        s3_source.connect().await.unwrap();

        // 获取存在的对象
        let content = s3_source.get_object("data/users.csv").unwrap();
        assert!(content.contains("id,name,email"));

        // 获取不存在的对象
        let result = s3_source.get_object("nonexistent.txt");
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_s3_execute_query() {
        let config = create_test_s3_config();
        let mut s3_source = S3DataSource::new(config);

        s3_source.connect().await.unwrap();

        // 执行列出对象的查询
        let query = DataSourceQuery {
            id: "test".to_string(),
            content: "SELECT * FROM objects".to_string(),
            parameters: HashMap::new(),
        };
        let result = s3_source.execute_query(query).await.unwrap();
        assert!(result.success);

        // 执行不支持的查询
        let query = DataSourceQuery {
            id: "test".to_string(),
            content: "INSERT INTO objects VALUES (...)".to_string(),
            parameters: HashMap::new(),
        };
        let result = s3_source.execute_query(query).await.unwrap();
        assert!(!result.success);
    }

    #[tokio::test]
    async fn test_s3_health_check() {
        let config = create_test_s3_config();
        let mut s3_source = S3DataSource::new(config);

        // 未连接时健康检查
        let health = s3_source.health_check().await.unwrap();
        assert_eq!(health["status"], "disconnected");

        // 连接后健康检查
        s3_source.connect().await.unwrap();
        let health = s3_source.health_check().await.unwrap();
        assert_eq!(health["status"], "connected");
    }

    #[tokio::test]
    async fn test_s3_query_without_connection() {
        let config = create_test_s3_config();
        let mut s3_source = S3DataSource::new(config);

        // 未连接时执行查询应该失败
        let query = DataSourceQuery {
            id: "test".to_string(),
            content: "SELECT * FROM objects".to_string(),
            parameters: HashMap::new(),
        };
        let result = s3_source.execute_query(query).await.unwrap();
        assert!(!result.success);
        assert!(result.error.is_some());
    }

    #[test]
    fn test_s3_config_default() {
        let config = S3Config::default();
        assert_eq!(config.endpoint, "https://s3.amazonaws.com");
        assert_eq!(config.region, "us-east-1");
        assert!(config.access_key_id.is_empty());
        assert!(config.secret_access_key.is_empty());
        assert!(config.bucket.is_empty());
        assert!(config.prefix.is_none());
    }

    #[test]
    fn test_s3_parse_sql() {
        let config = create_test_s3_config();
        let s3_source = S3DataSource::new(config);

        // 支持的查询
        assert_eq!(s3_source.parse_sql("SELECT * FROM objects").unwrap(), "list_objects");
        assert_eq!(s3_source.parse_sql("select key from s3").unwrap(), "list_objects");

        // 不支持的查询
        assert!(s3_source.parse_sql("INSERT INTO objects VALUES (...)").is_err());
        assert!(s3_source.parse_sql("SELECT * FROM users").is_err());
    }
}
